package com.kwan.xframe.mvp.model.api;

import android.util.Log;

import com.kwan.xframe.util.ByteArrayUtil;

import java.net.InetSocketAddress;
import java.net.Socket;

import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
import io.reactivex.FlowableEmitter;
import io.reactivex.FlowableOnSubscribe;
import okio.BufferedSink;
import okio.BufferedSource;
import okio.Okio;

public class RxSocket {

	private Socket socket;
	private BufferedSink mSink;
	private BufferedSource mSource;
	private String ip;
	private int port;

	//"192.168.0.26", 12000
	public RxSocket(String ip, int port) {
		this.ip = ip;
		this.port = port;
	}

	public Flowable<byte[]> write(final byte[] data) {
		return Flowable
				.create(new FlowableOnSubscribe<byte[]>() {
					@Override
					public void subscribe(FlowableEmitter<byte[]> e) throws Exception {

							if (socket != null && socket.isConnected()) {
								Log.e("kwan", "socket is connect now 2 close");
								socket.close();
							}

							socket = new Socket();
							socket.connect(new InetSocketAddress(ip, port));
							Log.e("kwan", "isConnected " + socket.isConnected());

							mSink = Okio.buffer(Okio.sink(socket));
							mSource = Okio.buffer(Okio.source(socket));

							mSink.write(data);
							mSink.flush();

							//此处是定制 处理
							// 第0个byte 是 请求号 第1个byte是数据长度

							byte[] buff = new byte[2];
							byte data[] = new byte[]{};

							for (int i = 0; i < 2; i++) {
								Log.e("kwan", "read -- " + mSource.read(buff));
								data = ByteArrayUtil.concatAll(data, buff);
							}

							//数据长度
							short slen = ByteArrayUtil.getShort(data, 2);

							Log.e("kwan", "data slen read -- " + slen);
							Log.e("kwan", "data read -- " + data.length);

							//根据数据长度 缓冲
							byte[] buff2 = new byte[slen];
							mSource.read(buff2);

							data = ByteArrayUtil.concatAll(data, buff2);
							//mSource.re
							Log.e("kwan", "data read -- " + data.length);

							e.onNext(data);
							e.onComplete();
							mSink.close();
							mSource.close();
							socket.close();

					}
				}, BackpressureStrategy.BUFFER);

	}
}
